package com.lagou.client.invoke;


import com.lagou.client.handler.LoadBalanceHandler;
import com.lagou.client.handler.RpcClientHandler;
import com.lagou.invoke.RpcRequest;
import com.lagou.invoke.RpcResponse;
import com.lagou.serializer.JSONSerializer;
import com.lagou.serializer.RpcDecoder;
import com.lagou.serializer.RpcEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Kirito
 * @description:
 * @date 2020/05/26
 */
public class RpcConsumer {

    private static final Logger log = LoggerFactory.getLogger(RpcConsumer.class);

    private static final ExecutorService executors = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private RpcClientHandler handler;

    private Bootstrap bootstrap;

    private Map<String, ChannelFuture> channels = new HashMap<>();

    public Object createProxy(final Class<?> clazz) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{clazz}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//                if (null == handler) {
                    initRpcClientHandler();
//                }

                RpcRequest request = new RpcRequest();
                request.setRequestId(UUID.randomUUID().toString().replaceAll("-", ""));
                request.setClassName(clazz.getName());
                request.setMethodName(method.getName());
                request.setParameters(args);
                if (null != args && args.length != 0) {
                    Class<?>[] parameterTypes = new Class[args.length];
                    for (int i = 0; i < parameterTypes.length; i++) {
                        parameterTypes[i] = args[i].getClass();
                    }
                    request.setParameterTypes(parameterTypes);
                }
                handler.setRequest(request);
                return executors.submit(handler).get();
            }
        });
    }

    public void initRpcClientHandler() {
        this.handler = new RpcClientHandler();
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(handler);
                    }
                });

        try {
//             for (String server : getServers()) {
//                 String[] address = StringUtils.split(server, ":");
//                 ChannelFuture future = this.bootstrap.connect(address[0], Integer.parseInt(address[1]));
//                 System.out.println("Connection established to " + server);
//                 channels.put(server, future);
//                 future.sync();
//             }
            LoadBalanceHandler loadBalance = new LoadBalanceHandler(getServers());
            String server = loadBalance.choose();
            String[] address = StringUtils.split(server, ":");
            ChannelFuture future = this.bootstrap.connect(address[0], Integer.parseInt(address[1]));
            log.info("Connection established to {}", server);
            channels.put(server, future);
            future.sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取服务端信息，即zk /servers下的子目录
     *
     * @return List
     */
    private List<String> getServers() {
        ZkClient zkClient = new ZkClient("192.168.73.128:2181");
        List<String> children = zkClient.getChildren("/servers");
        zkClient.subscribeChildChanges("/servers", ((s, list) -> {

            HashSet<String> after = new HashSet<>(list);
            HashSet<String> before = new HashSet<>(channels.keySet());
            // 如果变化前的服务不在变化后的列表里，那么就是需要断开连接的
            before.stream().filter(b -> !after.contains(b)).forEach(disConnect -> {
                ChannelFuture future = channels.get(disConnect);
                future.channel().close();
                log.info("DisConnect from {}", disConnect);

            });

            // 如果变化后的服务在缓存中不存在，那么建立新的连接
            after.stream().filter(a -> !before.contains(a)).forEach(newConnect -> {
                String[] address = StringUtils.split(newConnect, ":");
                ChannelFuture future = this.bootstrap.connect(address[0], Integer.parseInt(address[1]));
                channels.put(newConnect, future);
                log.info("Connection established to {}", newConnect);
                try {
                    future.sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }));
        return children;
    }

}
